package bigdata.backend.datasource.service.impl;

import com.baomidou.dynamic.datasource.DynamicRoutingDataSource;
import com.baomidou.dynamic.datasource.creator.DataSourceCreator;
import com.baomidou.dynamic.datasource.creator.DataSourceProperty;
import com.baomidou.dynamic.datasource.creator.hikaricp.HikariCpConfig;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import bigdata.backend.datasource.entity.Datasource;
import bigdata.backend.datasource.mapper.DatasourceMapper;
import bigdata.backend.datasource.service.DatasourceService;
import bigdata.backend.utils.DbType;
import bigdata.backend.datasource.utils.DdlBuilderUtils;
import bigdata.backend.utils.TableNameValidator;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.sql.DataSource;
import java.sql.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 数据源服务实现类
 * <p>
 * 职责：
 * - 动态注册/注销数据源到 DynamicRoutingDataSource
 * - 提供统一的 SQL 执行接口（查询、DML、DDL）
 * - 获取数据库元数据（表列表、字段信息、建表语句、注释等）
 * <p>
 * 设计原则：
 * - 线程安全：依赖 DynamicRoutingDataSource 内部同步机制
 * - 重试机制：对连接失效异常自动重试一次
 * - 类型安全：按 DbType 分发不同数据库的元数据查询逻辑
 * - 日志完备：关键操作均有 info/debug/error 日志
 */
@Slf4j
@Service
public class DataSourceServiceImpl extends ServiceImpl<DatasourceMapper, Datasource> implements DatasourceService {

    @Autowired
    private DynamicRoutingDataSource dynamicRoutingDataSource;

    @Autowired
    @Qualifier("hikariDataSourceCreator")
    private DataSourceCreator dataSourceCreator;


    private final Map<String, Object> registrationLocks = new ConcurrentHashMap<>();

    // ========== 数据源注册与连接管理 ==========

    /**
     * 动态注册数据源到路由容器
     *
     * @param ds 数据源配置实体
     * @return 创建的 DataSource 实例
     * @throws IllegalArgumentException 当数据源为空或别名为 'default' 时
     */
    @Override
    public DataSource registerDataSource(Datasource ds) {
        if (ds == null) {
            throw new IllegalArgumentException("Datasource must not be null");
        }
        String alias = generateAlias(ds);
        if ("default".equals(alias)) {
            throw new IllegalArgumentException("Alias 'default' is reserved for primary datasource and cannot be used");
        }

        log.debug("准备注册动态数据源: alias={}, host={}:{}, db={}",
                alias, ds.getHost(), ds.getPort(), ds.getDefaultDatabase());

        // 构建数据源属性
        DataSourceProperty property = buildDataSourceProperty(ds);

        // 创建并注册
        DataSource dataSource = dataSourceCreator.createDataSource(property);
        dynamicRoutingDataSource.addDataSource(alias, dataSource);

        log.info("动态数据源注册成功: alias={}", alias);
        return dataSource;
    }

    /**
     * 从路由容器中移除数据源
     *
     * @param alias 数据源别名
     * @throws IllegalArgumentException 当尝试移除 'default' 数据源时
     */
    @Override
    public void unregisterDataSource(String alias) {
        if ("default".equals(alias)) {
            throw new IllegalArgumentException("Cannot remove primary datasource");
        }
        // 获取与 getConnection 中相同的锁对象
        Object lock = registrationLocks.get(alias);
        if (lock != null) {
            synchronized (lock) {
                // 先检查是否存在（避免移除不存在的）
                if (dynamicRoutingDataSource.getDataSources().containsKey(alias)) {
                    dynamicRoutingDataSource.removeDataSource(alias);
                    log.info("动态数据源已移除: alias={}", alias);
                }
            }
        } else {
            // 没有锁，可能是预注册的，直接尝试移除（remove 内部是安全的）
            dynamicRoutingDataSource.removeDataSource(alias);
            log.debug("尝试移除不存在的数据源: alias={}", alias);
        }
    }

    /**
     * 获取数据源连接（带懒加载和重试）
     * <p>
     * 策略：
     * - 先查缓存
     * - 不存在则注册（幂等）
     * - 返回连接
     */
    private Connection getConnection(Datasource ds) throws SQLException {
        String alias = generateAlias(ds);
        // 第一次检查：直接查底层 map（不会抛异常）
        DataSource target = dynamicRoutingDataSource.getDataSources().get(alias);
        if (target == null) {
            // 检查 registrationLocks 中是否已有 alias 对应的锁对象
            // 如果没有：
            //    执行 k -> new Object()（即创建一个新的 Object 实例作为锁）
            //    将 alias → 新锁对象 存入 registrationLocks
            //    返回这个新锁对象
            // 如果已有：
            //    直接返回已存在的锁对象（不创建新对象）
            Object lock = registrationLocks.computeIfAbsent(alias, k -> new Object());
            synchronized (lock) {
                // 双重检查：仍然查 map，不是 getDataSource()
                target = dynamicRoutingDataSource.getDataSources().get(alias);
                if (target == null) {
                    target = registerDataSource(ds); // 直接拿到新创建的 DataSource
                }
            }
        }
        return target.getConnection();
    }

    /**
     * 将 ResultSet 转换为 List<Map>
     */
    private List<Map<String, Object>> resultSetToList(ResultSet rs) throws SQLException {
        List<Map<String, Object>> list = new ArrayList<>();
        ResultSetMetaData meta = rs.getMetaData();
        int cols = meta.getColumnCount();
        while (rs.next()) {
            HashMap<String, Object> row = new HashMap<>();
            for (int i = 1; i <= cols; i++) {
                row.put(meta.getColumnLabel(i), rs.getObject(i));
            }
            list.add(row);
        }
        return list;
    }

    // ========== SQL 执行核心逻辑 ==========

    /**
     * 执行 SQL（带连接失效重试）
     *
     * @param ds       数据源
     * @param sql      SQL 语句
     * @param params   参数（用于 PreparedStatement）
     * @param fetch    是否查询（返回 ResultSet）
     * @param isDml    是否 DML（返回影响行数）
     * @return 查询结果列表 或 DML 影响行数包装 或 null（DDL）
     */
    private List<Map<String, Object>> executeWithRetry(Datasource ds, String sql, Object[] params, boolean fetch, boolean isDml) {
        for (int attempt = 0; attempt <= 1; attempt++) { // 最多重试1次
            try (Connection conn = getConnection(ds)) {
                return executeSqlInternal(conn, sql, params, fetch, isDml);
            } catch (SQLException e) {
                if (isConnectionError(e) && attempt == 0) {
                    log.warn("检测到连接失效，尝试重新注册数据源并重试: {}", e.getMessage());
                    String alias = generateAlias(ds);
                    unregisterDataSource(alias);
                    continue; // 重试
                }
                log.error("SQL 执行失败 (attempt={})", attempt + 1, e);
                throw new RuntimeException("SQL 执行失败: " + e.getMessage(), e);
            }
        }
        return null; // 不可达
    }

    /**
     * 判断是否为连接相关异常（基于 JDBC 标准异常分类）
     */
    private boolean isConnectionError(SQLException e) {
        // 使用 SQLState 和异常类型判断，比字符串匹配更可靠
        return e instanceof SQLTransientConnectionException ||
                e instanceof SQLNonTransientConnectionException ||
                "08".equals(e.getSQLState()) || // SQLState 08xxx 表示连接异常
                (e.getErrorCode() == 0 && e.getMessage() != null &&
                        (e.getMessage().toLowerCase().contains("closed") ||
                                e.getMessage().toLowerCase().contains("gone away") ||
                                e.getMessage().toLowerCase().contains("broken pipe") ||
                                e.getMessage().toLowerCase().contains("connection reset")));
    }

    /**
     * 在已有连接上执行 SQL
     */
    private List<Map<String, Object>> executeSqlInternal(Connection conn, String sql, Object[] params, boolean fetch, boolean isDml)
            throws SQLException {
        try (PreparedStatement ps = conn.prepareStatement(sql)) {
            if (params != null) {
                for (int i = 0; i < params.length; i++) {
                    ps.setObject(i + 1, params[i]);
                }
            }
            if (fetch) {
                // 查询语句（SELECT / SHOW 等），期望返回 ResultSet
                try (ResultSet rs = ps.executeQuery()) {
                    return resultSetToList(rs);
                }
            } else if (isDml) {
                // DML 语句（INSERT/UPDATE/DELETE），使用 executeUpdate() 并返回影响行数
                // 注意：此处 isDml 表示“需要返回 rowcount”，语义上更偏向“是否为数据变更操作”
                int count = ps.executeUpdate();
                return Arrays.asList(Collections.singletonMap("rowcount", count));
            } else {
                // DDL 或其他无结果集、无需返回行数的语句（如 CREATE/ALTER/DROP/TRUNCATE/COMMENT）
                // 使用 execute() 执行，不关心返回值
                ps.execute();
                return null;
            }
        }
    }

    // ========== 元数据查询 ==========

    @Override
    public List<String> getAllTables(Datasource ds) {
        DbType type = DbType.fromString(ds.getDbType());
        if (type == DbType.MYSQL || type == DbType.DORIS) {
            List<Map<String, Object>> rows = executeWithRetry(ds, "SHOW TABLES", null, true, false);
            String key = "Tables_in_" + ds.getDefaultDatabase();
            return rows.stream().map(r -> (String) r.get(key)).collect(Collectors.toList());
        } else if (type == DbType.POSTGRESQL) {
            String schema = ds.getEffectSchema();
            List<Map<String, Object>> rows = executeQuery(ds,
                    "SELECT table_name FROM information_schema.tables WHERE table_schema = ? AND table_type = 'BASE TABLE'",
                    schema);
            return rows.stream().map(r -> (String) r.get("table_name")).collect(Collectors.toList());
        }
        throw new UnsupportedOperationException("不支持的数据库类型");
    }

    @Override
    public List<Map<String, Object>> getTableColumns(Datasource ds, String tableName) {
        tableName = TableNameValidator.validate(tableName);
        DbType type = DbType.fromString(ds.getDbType());
        if (type == DbType.MYSQL || type == DbType.DORIS) {
            return getTableColumnsFromMysqlOrDoris(ds, tableName);
        } else if (type == DbType.POSTGRESQL) {
            return getTableColumnsFromPostgreSQL(ds, tableName);
        }
        throw new UnsupportedOperationException("不支持的数据库类型");
    }

    /**
     * 从 MySQL 或 Doris 获取表字段信息
     */
    private List<Map<String, Object>> getTableColumnsFromMysqlOrDoris(Datasource ds, String tableName) {
        List<Map<String, Object>> rows = executeQuery(ds,
                "SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLUMN_COMMENT, EXTRA, COLUMN_KEY " +
                        "FROM information_schema.COLUMNS " +
                        "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? ORDER BY ORDINAL_POSITION",
                ds.getDefaultDatabase(), tableName);

        return rows.stream().map(row -> {
            Map<String, Object> col = new HashMap<>();
            col.put("name", row.get("COLUMN_NAME"));
            col.put("type", row.get("COLUMN_TYPE"));
            col.put("nullable", "YES".equals(row.get("IS_NULLABLE")));
            col.put("default", row.get("COLUMN_DEFAULT"));
            col.put("comment", row.get("COLUMN_COMMENT") != null ? row.get("COLUMN_COMMENT") : "");
            col.put("extra", row.get("EXTRA") != null ? row.get("EXTRA") : "");
            col.put("is_primary_key", "PRI".equals(row.get("COLUMN_KEY")));
            return col;
        }).collect(Collectors.toList());
    }

    /**
     * 从 PostgreSQL 获取表字段信息（使用参数化查询减少拼接）
     */
    private List<Map<String, Object>> getTableColumnsFromPostgreSQL(Datasource ds, String tableName) {
        String schema = ds.getEffectSchema();

        String sql = String.format(
                "SELECT  " +
                        "c.column_name, " +
                        "c.data_type, " +
                        "c.character_maximum_length, " +
                        "c.numeric_precision, " +
                        "c.numeric_scale, " +
                        "c.is_nullable, " +
                        "c.column_default, " +
                        "pg_catalog.col_description( " +
                        "(SELECT oid FROM pg_class WHERE relname = '%s' AND relnamespace =  " +
                        "(SELECT oid FROM pg_namespace WHERE nspname = '%s') " +
                        "), c.ordinal_position " +
                        ") AS column_comment, " +
                        "CASE  " +
                        "WHEN pk_cols.column_name IS NOT NULL THEN TRUE  " +
                        "ELSE FALSE  " +
                        "END AS is_primary_key " +
                        "FROM information_schema.columns c " +
                        "LEFT JOIN ( " +
                        "SELECT kcu.column_name " +
                        "FROM information_schema.table_constraints tc " +
                        "JOIN information_schema.key_column_usage kcu " +
                        "ON tc.constraint_name = kcu.constraint_name " +
                        "AND tc.table_schema = kcu.table_schema " +
                        "WHERE tc.constraint_type = 'PRIMARY KEY' " +
                        "AND tc.table_schema = '%s' " +
                        "AND tc.table_name = '%s' " +
                        ") pk_cols ON c.column_name = pk_cols.column_name " +
                        "WHERE c.table_schema = '%s' AND c.table_name = '%s' " +
                        "ORDER BY c.ordinal_position ",
                tableName, schema, schema, tableName, schema, tableName
        );
        List<Map<String, Object>> rows = executeQuery(ds, sql);

        return rows.stream().map(row -> {
            String dataType = (String) row.get("data_type");
            Integer charLen = (Integer) row.get("character_maximum_length");
            String displayType;

            // 类型标准化
            if ("character varying".equals(dataType)) {
                displayType = charLen != null ? "varchar(" + charLen + ")" : "varchar";
            } else if ("character".equals(dataType)) {
                displayType = charLen != null ? "char(" + charLen + ")" : "char";
            } else if ("numeric".equals(dataType)) {
                Integer prec = (Integer) row.get("numeric_precision");
                Integer scale = (Integer) row.get("numeric_scale");
                displayType = (prec != null && scale != null) ? "numeric(" + prec + "," + scale + ")" : "numeric";
            } else {
                switch (dataType) {
                    case "integer":
                        displayType = "int";
                        break;
                    case "bigint":
                        displayType = "bigint";
                        break;
                    case "double precision":
                        displayType = "double";
                        break;
                    case "timestamp without time zone":
                        displayType = "timestamp";
                        break;
                    default:
                        displayType = dataType;
                        break;
                }
            }

            Map<String, Object> col = new HashMap<>();
            col.put("name", row.get("column_name"));
            col.put("type", displayType);
            col.put("nullable", "YES".equals(row.get("is_nullable")));
            // pg默认值可能包含::需要特殊处理
            Object rawDefault = row.get("column_default");
            if (rawDefault != null) {
                String columnDefault = rawDefault.toString();
                // 1. 去掉 ::type
                if (columnDefault.contains("::")) {
                    columnDefault = columnDefault.split("::", 2)[0]; // 只 split 一次，避免内容中有 ::
                }
                // 2. 如果是字符串字面量（以 ' 开头和结尾），去掉外层引号并处理转义
                if (columnDefault.length() >= 2 &&
                        columnDefault.startsWith("'") &&
                        columnDefault.endsWith("'")) {
                    // 去掉首尾单引号
                    columnDefault = columnDefault.substring(1, columnDefault.length() - 1);
                    // 将 '' 替换为 '
                    columnDefault = columnDefault.replace("''", "'");
                }
                col.put("default", columnDefault);
            } else {
                col.put("default", null);
            }
            col.put("comment", row.get("column_comment") != null ? row.get("column_comment") : "");
            col.put("extra", "");
            col.put("is_primary_key", row.get("is_primary_key"));
            return col;
        }).collect(Collectors.toList());
    }

    @Override
    public String getCreateTableSql(Datasource ds, String tableName) {
        tableName = TableNameValidator.validate(tableName);
        DbType type = DbType.fromString(ds.getDbType());
        if (type == DbType.MYSQL || type == DbType.DORIS) {
            List<Map<String, Object>> rows = executeWithRetry(ds, "SHOW CREATE TABLE `" + tableName + "`", new Object[0], true, false);
            return (String) rows.get(0).get("Create Table");
        } else if (type == DbType.POSTGRESQL) {
            // PostgreSQL 无原生 SHOW CREATE TABLE，需重建
            List<Map<String, Object>> columns = getTableColumns(ds, tableName);
            String tableComment = getTableComment(ds, tableName);
            String schema = ds.getEffectSchema();
            String createSql = DdlBuilderUtils.buildCreateTableSql(
                    columns, type, type, tableName, schema, tableComment, true);
            return createSql;
        }
        throw new UnsupportedOperationException("不支持的数据库类型");
    }

    @Override
    public String getTableComment(Datasource ds, String tableName) {
        tableName = TableNameValidator.validate(tableName);
        DbType type = DbType.fromString(ds.getDbType());
        if (type == DbType.MYSQL || type == DbType.DORIS) {
            List<Map<String, Object>> rows = executeQuery(ds,
                    "SELECT TABLE_COMMENT FROM information_schema.TABLES WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?",
                    ds.getDefaultDatabase(), tableName);
            return rows.isEmpty() ? null : (String) rows.get(0).get("TABLE_COMMENT");
        } else if (type == DbType.POSTGRESQL) {
            String schema = ds.getEffectSchema();
            List<Map<String, Object>> rows = executeQuery(ds,
                    "SELECT obj_description(c.oid, 'pg_class') AS table_comment " +
                            "FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace " +
                            "WHERE c.relname = ? AND n.nspname = ?",
                    tableName, schema);
            return rows.isEmpty() ? null : (String) rows.get(0).get("table_comment");
        }
        return null;
    }

    // ========== SQL 执行公开接口 ==========

    @Override
    public List<Map<String, Object>> executeQuery(Datasource ds, String sql) {
        return executeQuery(ds, sql, new Object[0]);
    }

    @Override
    public List<Map<String, Object>> executeQuery(Datasource ds, String sql, Object... params) {
        if (sql.trim().toUpperCase().startsWith("SELECT") || sql.trim().toUpperCase().startsWith("SHOW")) {
            return executeWithRetry(ds, sql, params, true, false);
        } else {
            throw new IllegalArgumentException("仅支持 SELECT、SHOW 查询");
        }
    }

    @Override
    public int executeUpdate(Datasource ds, String sql) {
        return executeUpdate(ds, sql, new Object[0]);
    }

    @Override
    public int executeUpdate(Datasource ds, String sql, Object... params) {
        String upper = sql.trim().toUpperCase();
        if (!(upper.startsWith("INSERT") || upper.startsWith("UPDATE") || upper.startsWith("DELETE"))) {
            throw new IllegalArgumentException("仅支持 DML 语句");
        }
        List<Map<String, Object>> result = executeWithRetry(ds, sql, params, false, true);
        return result == null ? 0 : (int) result.get(0).get("rowcount");
    }

    @Override
    public void executeDdl(Datasource ds, String sql) {
        executeDdl(ds, sql, new Object[0]);
    }

    @Override
    public void executeDdl(Datasource ds, String sql, Object... params) {
        for (String stmt : splitSql(sql)) {
            String upper = stmt.trim().toUpperCase();
            if (!(upper.startsWith("CREATE") || upper.startsWith("ALTER") ||
                    upper.startsWith("DROP") || upper.startsWith("TRUNCATE") || upper.startsWith("COMMENT"))) {
                throw new IllegalArgumentException("DDL 仅支持 CREATE/ALTER/DROP/TRUNCATE/COMMENT");
            }
            executeWithRetry(ds, stmt, params, false, false);
        }
    }

    @Override
    public boolean testConnection(Datasource ds) {
        try (Connection conn = getConnection(ds)) {
            return conn != null && !conn.isClosed();
        } catch (Exception e) {
            log.warn("数据源连接测试失败: alias={}, error={}",
                    ds.getDbAlias() != null ? ds.getDbAlias() : generateAlias(ds),
                    e.getMessage());
            return false;
        }
    }

    @Override
    public boolean testConnectionById(String id) {
        Datasource ds = this.getById(id);
        if (ds == null) {
            log.warn("数据源不存在: id={}", id);
            return false;
        }
        // 直接调用带重试的 getConnection 尝试获取连接
        try (Connection conn = getConnection(ds)) {
            return conn != null && !conn.isClosed();
        } catch (Exception e) {
            log.warn("测试数据源连接失败: id={}, alias={}, error={}",
                    id, generateAlias(ds), e.getMessage());
            return false;
        }
    }

    @Override
    @Transactional
    public boolean updateById(Datasource datasource) {
        // 1. 先查旧数据
        Datasource old = getById(datasource.getId());
        if (old == null) {
            throw new RuntimeException("数据源不存在");
        }

        // 2. 执行更新
        boolean updated = super.updateById(datasource);

        // 3. 如果关键字段变更（host/port/user/pass/db），则注销旧连接
        if (isConnectionConfigChanged(old, datasource)) {
            String alias = generateAlias(old); // 用旧配置生成 alias（因为旧连接是用旧配置注册的）
            try {
                unregisterDataSource(alias);
                log.info("检测到连接信息变更，已注销旧数据源连接: alias={}", alias);
            } catch (Exception e) {
                log.warn("注销数据源连接失败，可能尚未注册: alias={}", alias, e);
            }
        }

        return updated;
    }

    // MyBatis-Plus 的默认 updateById() 会忽略值为 null 的字段，不会生成 SET column = NULL
    @Transactional
    public boolean updateByIdIncludingNulls(Datasource datasource) {

        Datasource existing = getById(datasource.getId());
        if (existing == null) {
            throw new RuntimeException("数据源不存在");
        }

        LambdaUpdateWrapper<Datasource> update = new LambdaUpdateWrapper<>();
        update.eq(Datasource::getId, datasource.getId());

        // 显式设置每个字段（包括 null）
        update.set(Datasource::getSourceName, datasource.getSourceName());
        update.set(Datasource::getHost, datasource.getHost());
        update.set(Datasource::getUsername, datasource.getUsername());
        update.set(Datasource::getPassword, datasource.getPassword());
        update.set(Datasource::getPort, datasource.getPort());
        update.set(Datasource::getDefaultDatabase, datasource.getDefaultDatabase());
        update.set(Datasource::getSchemaName, datasource.getSchemaName()); // null 也会被设为 NULL
        update.set(Datasource::getDbAlias, datasource.getDbAlias());
        update.set(Datasource::getDbType, datasource.getDbType());
        // 注意：不要更新 createTime/updateTime/deleteFlag

        boolean updated = update(new Datasource(), update);

        // 检查连接配置是否变更（包括 schemaName！）
        if (updated && isConnectionConfigChanged(existing, datasource)) {
            String oldAlias = generateAlias(existing);
            unregisterDataSource(oldAlias);
        }

        return updated;
    }

    /**
     * 判断连接关键配置是否变更
     */
    private boolean isConnectionConfigChanged(Datasource old, Datasource now) {
        return !Objects.equals(old.getHost(), now.getHost()) ||
                !Objects.equals(old.getPort(), now.getPort()) ||
                !Objects.equals(old.getUsername(), now.getUsername()) ||
                !Objects.equals(old.getPassword(), now.getPassword()) ||
                !Objects.equals(old.getDefaultDatabase(), now.getDefaultDatabase()) ||
                !Objects.equals(old.getSchemaName(), now.getSchemaName()) ||
                !Objects.equals(old.getDbType(), now.getDbType());
    }

    private List<String> splitSql(String sql) {
        return Arrays.stream(sql.split(";"))
                .map(String::trim)
                .filter(s -> !s.isEmpty())
                .collect(Collectors.toList());
    }

    // ========== 工具方法 ==========

    /**
     * 生成数据源唯一别名（基于连接信息 MD5）
     */
    private String generateAlias(Datasource ds) {
        String schema = ds.getEffectSchema();

        String fingerprintInput = String.format(
                "%s:%d/%s/%s?user=%s&pass=%s",
                ds.getHost(),
                ds.getPort(),
                Objects.toString(ds.getDefaultDatabase(), ""),
                Objects.toString(schema, ""),
                Objects.toString(ds.getUsername(), ""),
                Objects.toString(ds.getPassword(), "")
        );
        String fingerprint = DigestUtils.md5Hex(fingerprintInput);
        return "dynamic_" + fingerprint.substring(0, 8);
    }

    /**
     * 构建 JDBC URL
     */
    private String buildJdbcUrl(Datasource ds) {
        DbType dbType = DbType.fromString(ds.getDbType());
        String host = ds.getHost();
        int port = ds.getPort();
        String db = ds.getDefaultDatabase();
        switch (dbType) {
            case MYSQL:
            case DORIS:
                return "jdbc:mysql://" + host + ":" + port + "/" + db +
                        "?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true&useSSL=false";
            case POSTGRESQL:
                return "jdbc:postgresql://" + host + ":" + port + "/" + db;
            default:
                throw new IllegalArgumentException("不支持的类型: " + dbType);
        }
    }

    /**
     * 获取 JDBC 驱动类名
     */
    public String getDriverClass(String dbType) {
        switch (DbType.fromString(dbType)) {
            case MYSQL:
            case DORIS:
                return "com.mysql.cj.jdbc.Driver";
            case POSTGRESQL:
                return "org.postgresql.Driver";
            default:
                throw new IllegalArgumentException("不支持的类型: " + dbType);
        }
    }

    /**
     * 构建 DataSourceProperty（含连接池配置）
     */
    private DataSourceProperty buildDataSourceProperty(Datasource ds) {
        DataSourceProperty property = new DataSourceProperty();
        // 设置基础连接信息
        property.setUsername(ds.getUsername());
        property.setPassword(ds.getPassword());
        // 构建JDBC URL
        property.setUrl(buildJdbcUrl(ds));
        // 设置驱动类
        property.setDriverClassName(getDriverClass(ds.getDbType()));
        // 可选 设置连接池参数（HikariCP）
        property.setHikari(buildHikariConfig(ds));
        return property;
    }

    /**
     * 构建 HikariCP 连接池配置
     * <p>
     * TODO: 未来可从配置中心读取
     */
    private HikariCpConfig buildHikariConfig(Datasource ds) {
        HikariCpConfig config = new HikariCpConfig();
        config.setConnectionTimeout(30000L);   // 30秒
        config.setIdleTimeout(600000L);        // 10分钟
        config.setMaxLifetime(1800000L);       // 30分钟
        config.setMaximumPoolSize(20);
        config.setMinIdle(5);
        return config;
    }
}
